import scala.actors._
import scala.actors.Actor._
import scala.math
import scala.util.Random

import project2._

object Rumor
object Converge
object Start
object Stop
object Termination // to peers
case class State(s:Double, w:Double)
case class Result(s:Double, w:Double)
case class Flood(converge: Long)

class Topology(numNode: Int, proto: String) extends Actor{

	var allNodes: Array[Node] = null
	var n = numNode
	var runningNode = n
    val detectThreshold = 0.5
    val protocolType = proto
    var startTime = 0.toLong
	
	def act() {

        if(protocolType == "gossip"){
            startTime = System.currentTimeMillis
            allNodes(new Random().nextInt(n)) ! Rumor
            loop{
                react {
                    case Converge => {
                        runningNode -= 1

                        //
                        // with termination detection
                        //

                        if(runningNode == 0){
                            println(System.currentTimeMillis-startTime + " ms")
                            exit()
                        }
                    }
                }
            } // loop
        } else if(protocolType == "push-sum"){
            var sum = 0.toDouble
            var weight = 0.toDouble
            if(DEBUG)
                println("master started")
            loop{
                react {
                    case Result(s,w) => {
                        sum += s
                        weight += w
                        runningNode -= 1
                        if(DEBUG == true){
                            println("sum is " + s/w)
                        }

                        //
                        // with termination detection
                        //
                        if(runningNode < n * detectThreshold){
                            println(System.currentTimeMillis-startTime)
                            killAll()
                            exit()
                        }
                    }
                }
            } // loop }
        }
	}

    def killAll(){
        allNodes.foreach(e=>e!Stop)
    }
	
	def printTopology(){
	    for(i <- 0 until allNodes.length){
	        val node = allNodes(i)
	        print(node.id + ": ")
	        for(j <- 0 until node.neighborList.length){
	            print(node.neighborList(j).id + " ")
	        }
	        println("")
	    }
	}
	
	def createNodes(num: Int, proto: String, conn: Int){
		allNodes = new Array[Node](num)
		for(i <- 0 until num){
		    allNodes(i) = new Node(i, this, proto, conn)
		    allNodes(i).id = i
		}
	}
	
	def createFull(proto: String){
	    createNodes(n, proto, n-1)
	    allNodes.foreach( e => e.addNodes(allNodes.filter(x => x.id!=e.id)) )
	}
	def create2D(proto: String, conn: Int){
	    //println("2D")
	    val side = math.ceil(math.sqrt(n)).toInt
	    n = math.pow(side, 2).toInt
	    createNodes(n, proto, conn)
		for(i <- 0 until side){
		    for(j <- 0 until side){
		        // get index of node
		        val index = i*side + j
			    if(i == 0){ // first line
			        if(j == 0){ // first column
			            // add two nodes to list
			            val neighbors = new Array[Node](2)
			            neighbors(0) = allNodes(index+side)	// down
			            neighbors(1) = allNodes(index+1)	// right
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        } else if(j == side - 1){ // last column
			            // add two nodes to list
			            val neighbors = new Array[Node](2)
			            neighbors(0) = allNodes(index+side)	// down
			            neighbors(1) = allNodes(index-1)	// left
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        } else{ // middle column
			            // add three nodes to list
			            val neighbors = new Array[Node](3)
			            neighbors(0) = allNodes(index-1)	// left
			            neighbors(1) = allNodes(index+1)	// right
			            neighbors(2) = allNodes(index+side)	// down
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        }
			    } else if(i == side - 1){ // if the last line
			        if(j == 0){ // first column
			            // add two nodes to list
			            val neighbors = new Array[Node](2)
			            neighbors(0) = allNodes(index-side)	// up
			            neighbors(1) = allNodes(index+1)	// right
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        } else if(j == side - 1){ // last column
			            // add two nodes to list
			            val neighbors = new Array[Node](2)
			            neighbors(0) = allNodes(index-side)	// up
			            neighbors(1) = allNodes(index-1)	// left
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        } else{ // middle column
			            // add three nodes to list
			            val neighbors = new Array[Node](3)
			            neighbors(0) = allNodes(index-1)	// left
			            neighbors(1) = allNodes(index+1)	// right
			            neighbors(2) = allNodes(index-side)	// up
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        }
			    } else { // normal line
			        if(j == 0){ // first column
			            // add two nodes to list
			            val neighbors = new Array[Node](3)
			            neighbors(0) = allNodes(index-side)	// up
			            neighbors(1) = allNodes(index+1)	// right
			            neighbors(2) = allNodes(index+side)	// down
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        } else if(j == side - 1){ // last column
			            // add two nodes to list
			            val neighbors = new Array[Node](3)
			            neighbors(0) = allNodes(index-side)	// up
			            neighbors(1) = allNodes(index-1)	// left
			            neighbors(2) = allNodes(index+side)	// down
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        } else{ // middle column
			            // add four nodes to list
			            val neighbors = new Array[Node](4)
			            neighbors(0) = allNodes(index-1)	// left
			            neighbors(1) = allNodes(index+1)	// right
			            neighbors(2) = allNodes(index-side)	// up
			            neighbors(3) = allNodes(index+side)	// down
			            // add list to node
			            allNodes(index).addNodes(neighbors)
			        }
			    }
		    } //for j
		} //for i
	}
	def createLine(proto: String){
	    createNodes(n, proto, 2)
	    for(i <- 0 until n){
            if(i == 0){
                allNodes(i).addNodes(Array(allNodes(i+1)))
            } else if(i == n-1){
                allNodes(i).addNodes(Array(allNodes(i-1)))
            } else{
                allNodes(i).addNodes(Array(allNodes(i-1), allNodes(i+1)))
            }
	    }
	}
	def createImp2D(proto: String){
	    create2D(proto, 5)
        addOneRandomLink()
	}
	def addOneRandomLink(){
	    var randomId = 0
        val gen = new Random()
        val added = Array.fill(n){false}
        var i = 0
        for(i <- 0 until n){
            if(added(i) == false){
                val candidates = (0 until n toArray).filter(x => added(x)==false & x != i & isNeighbor(i, x) == false)
                if(candidates.length != 0){
                    randomId = candidates(gen.nextInt(candidates.length))
                    allNodes(i).appendNodes(Array(allNodes(randomId)))
                    allNodes(randomId).appendNodes(Array(allNodes(i)))
                    added(i) = true
                    added(randomId) = true
                } else {
                    added(i) = true
                }
            }
        }
	}

    def isNeighbor(i: Int, j: Int): Boolean = {
        val list = allNodes(i).neighborList
        for(nei <- list){
            if(nei.id == j){
                return true
            }
        }
        return false
    }
	
	def startAll(){
	    allNodes.foreach( node => node.start() )
	}
	def gossip(graphType:String){
        buildGraph(graphType, "gossip")
        if(DEBUG == true)
            printTopology()
	    startAll()
	    this.start()
	}
	def pushSum(graphType:String){
        if(DEBUG == true)
            println("push-sum")
        buildGraph(graphType, "push-sum")
        if(DEBUG == true)
            printTopology()
	    startAll()
        if(DEBUG == true)
            println("all actor started")
        this.start()
        flood()
	}

    def flood(){
        val trueSum = ((n-1)*n/2).toLong
        startTime = System.currentTimeMillis
        val queryPt = new Random().nextInt(n)
        for(node <- allNodes){
            node ! Flood(trueSum)
            if(node.id == queryPt){
                for(nn <- allNodes){
                    nn ! State(node.id.toDouble/n, 1.toDouble/n)
                }
            }
            else
                node ! State(node.id, 0)
        }
        if(DEBUG == true)
            println("flooded")
    }

    def buildGraph(graphType:String, proto: String){
		if(graphType == "full"){
			createFull(proto)
			//graph.printTopology()
		} else if (graphType == "2D"){
			create2D(proto, 4)
			//graph.printTopology()
		} else if (graphType == "line"){
			createLine(proto)
		    //graph.printTopology()
		} else if (graphType == "imp2D"){
			createImp2D(proto)
			//graph.printTopology()
		}
    }
}

class Node(nid: Int, masterRef: Topology, protocol: String, conn: Int) extends Actor {
    var neighborList: Array[Node] = null
    var id = nid
    val gen = new Random()
    var rumorReceived = 0
    val master = masterRef
    val terminationCond = 10
    var isTerminated = false
    val protocolType = protocol
    var activeNeighborNum = 0
    val connectivity = conn

    def addNodes(neighbors: Array[Node]){
        neighborList = neighbors
    }
    def appendNodes(list: Array[Node]){
        neighborList = neighborList ++ list
    }
    def chooseRandomNeighbor(): Node = {
        if(neighborList.length == 0){
            return null
        }
        return neighborList(gen.nextInt(neighborList.length))
    }
    def chooseRandomActiveNeighbor(): Node = {
        if(neighborList.length == 0){
            return null
        }
        var ret: Node = null
        var idx = -1
        do{
            idx = gen.nextInt(neighborList.length)
        } while(neighborList(idx).isTerminated == true)
        return neighborList(idx)
    }

    def act = {
        if(protocolType == "gossip") {
            activeNeighborNum = neighborList.length
            loop{
                reactWithin(2){
                    case Rumor => {
                        // if still active, process Rumor, else just igonor it
                        if(isTerminated == false){
                            if(rumorReceived == terminationCond){
                                isTerminated = true
                                master ! Converge
                                for(neighbor <- neighborList){
                                    neighbor ! Termination
                                }
                                if(DEBUG == true)
                                    println(id + " is in termination")
                                exit()
                            } else {
                                rumorReceived += 1
                                if(DEBUG == true)
                                    println(id + " receives " + rumorReceived + " msg")
                            }
                        }
                    } 
                    // Termination detection
                    case Termination => {
                        activeNeighborNum -= 1
                        if(isTerminated == false & activeNeighborNum <= 0){
                            master ! Converge
                            exit()
                        }
                    }
                    case Stop => {
                        exit()
                    }
                    case TIMEOUT => {
                        if(0 < rumorReceived & rumorReceived <= terminationCond){
                            val rec = chooseRandomNeighbor() 
                            ///println(id + " is sending to " + rec.id)
                            rec ! Rumor
                        }
                    }
                }
            } // loop
        } // end of if run gossip proto
        //
        // if run push-sum proto
        //
        else if(protocolType == "push-sum"){
            var evaluation = -1.toDouble
            var old_evaluation = -1.toDouble
            val threshold = 0.25
            var satisfyTime = 0
            var sum = 0.toDouble
            var weight = 0.toDouble
            var activeNeighborNum = neighborList.length
            var inited = false
            isTerminated = false
            val ROUND = 50.toLong
            var waitTime = ROUND
            var lastTime = System.currentTimeMillis
            var convergeAt = 0.toLong


            loop{
                react {
                    case State(s,w) => {
                        if(isTerminated == false){
                            sum += s
                            weight += w

                            evaluation = sum/weight
                            old_evaluation = evaluation
                            if(DEBUG)
                                println(id + " evaluation " + evaluation)

                            // random communication
                            val rec = chooseRandomNeighbor()
                            rec ! State(sum/2, weight/2)
                            if(DEBUG == true){
                                println(id + " is sending " + (sum/2,weight/2) + " to " + rec.id)
                            }

                            if(math.abs(evaluation-convergeAt) < threshold * convergeAt){
                            //if(math.abs(evaluation-old_evaluation) < threshold){
                                satisfyTime += 1
                                if(satisfyTime >= 3){
                                    isTerminated = true
                                    master ! Result(sum, weight)
                                    if(DEBUG == true){
                                        println(id + " is terminating")
                                    }
                                }
                            }

                            //send the other half to self
                            sum = sum/2
                            weight = weight/2
                        } else{
                            // just be a router    
                            val rec = chooseRandomNeighbor()
                            rec ! State(s, w)
                        }
                    }
                    case Flood(s) =>{
                        convergeAt = s
                    }
                    case Stop => {
                        exit()    
                    }
                }
            }
            /* round version
            loop{
                // round time is 2
                reactWithin(waitTime) {
                    case State(s,w) => {
                        sum += s
                        weight += w
                        if(DEBUG == true){
                            println(id + " receiving " + (s, w))
                        }
                        waitTime -= (System.currentTimeMillis - lastTime)
                        if(waitTime < 0)
                            waitTime = 0
                        lastTime = System.currentTimeMillis
                    }
                    case Flood(s) =>{
                        convergeAt = s
                    }
                    case TIMEOUT => {
                        // keep evaluation
                        old_evaluation = evaluation
                        evaluation = sum/weight

                        if(math.abs(evaluation-old_evaluation) < threshold){
                            satisfyTime += 1
                            if(satisfyTime >= 3){
                                isTerminated = true
                                master ! Result(sum, weight)
                                if(DEBUG == true){
                                    println(id + " is terminating")
                                }
                                exit()
                            }
                        }

                        // clean slate
                        waitTime = ROUND
                        lastTime = System.currentTimeMillis

                        // random gossip
                        val rec = chooseRandomNeighbor()
                        
                        if(DEBUG == true){
                            println(id + " is sending to " + (sum/2,weight/2) + " to self and " + rec.id)
                        }
                        rec ! State(sum/2, weight/2)
                        self ! State(sum/2, weight/2)

                        sum = 0.toDouble
                        weight = 0.toDouble
                    }
                }
            }
            */
            /*
            loop{
                // round time is 2
                reactWithin(5) {
                    case State(s,w) => {
                        transient = 0
                        // accumulate
                        if(DEBUG == true){
                            println(id + " is receiving " + (s,w))
                        }
                        sum = sum + s
                        weight = weight + w

                        // 1. keep old evaluation
                        // 2. get new evaluation
                        if(inited == false){
                            inited = true
                            old_evaluation = sum/weight
                        } else{
                            old_evaluation = evaluation
                        }
                        evaluation = sum/weight

                        // termination detection
                        if(math.abs(evaluation-old_evaluation) < threshold){
                            satisfyTime += 1
                            if(satisfyTime >= 3){
                                isTerminated = true
                                for(neighbor <- neighborList){
                                    if(DEBUG == true){
                                        println(id + " is sending Termination to " + neighbor.id)
                                    }
                                    neighbor ! Termination
                                }
                                master ! Result(sum, weight)
                                if(DEBUG == true){
                                    println(id + " is terminating")
                                }
                                exit()
                            }
                        } else {
                            satisfyTime = 0
                        }

                        // random communication
                        val rec = chooseRandomNeighbor()
                        rec ! State(sum/2, weight/2)
                        if(DEBUG == true){
                            println(id + " is sending " + (sum/2,weight/2) + " to " + rec.id)
                        }
                        //send the other half to self
                        sum = sum/2
                        weight = weight/2
                    }
                    case Termination => {
                        activeNeighborNum -= 1
                        if(DEBUG == true){
                            println(id + " receives termination, left = " + activeNeighborNum)
                        }
                        if(isTerminated == false & activeNeighborNum <= 0){
                            isTerminated = true
                            master ! Result(sum, weight)
                            exit()
                        }
                    }
                    case TIMEOUT => {
                        transient += 1
                        if(transient >= 10){
                            isTerminated = true
                            for(neighbor <- neighborList){
                                if(DEBUG == true){
                                    println(id + " is sending Termination to " + neighbor.id)
                                }
                                neighbor ! Termination
                            }
                            master ! Result(sum, weight)
                            if(DEBUG == true){
                                println(id + " is terminating")
                            }
                            exit()
                        }
                    }
                } // reactWithin
            }//loop
            */
        }//push-sum
    }
}

object project2 {
    val DEBUG=false
	def main(args: Array[String]) = {
		val graphType = args(1).toString
		val algo = args(2).toString
		val graph = new Topology(args(0).toInt, algo)
		
		if(algo == "gossip"){
		    graph.gossip(graphType)
		} else if(algo == "push-sum"){
		    graph.pushSum(graphType)
		}
	}
}
